package skewreduce.framework.planner.axis;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

import skewreduce.framework.planner.EaggCostModel;
import skewreduce.framework.planner.PartitionNode;
import skewreduce.lib.ITuple;
import skewreduce.lib.Partition;


public abstract class AxisChooser {
	public abstract Axis choose(PartitionNode p, Axis... axes);
	
	public static final class PreparedAxes {
		Axis[] axes;
		int[]  sortOrder;
		private PreparedAxes(List<Axis> axes,List<Axis> sort) {
			this.axes = axes.toArray(new Axis[axes.size()]);
			sortOrder = new int[sort.size()];
			for ( int i = 0; i < sortOrder.length; ++i )
				sortOrder[i] = sort.get(i).getIndex();
		}
		
		public Axis[] getAxes() { return axes; }
		public int[] getSortOrder() { return sortOrder; }
	}
	
	public static PreparedAxes prepareAxes(EaggCostModel model,PartitionNode node) {
		Partition partition = node.getPartition();
		ArrayList<Axis> tmp = new ArrayList<Axis>(partition.getArity());
		ArrayList<Axis> sortOrder = new ArrayList<Axis>(partition.getArity());
		for ( int i = 0; i < partition.getArity(); ++i ) {
			Axis a = new Axis(partition,i);
			model.initAxis(partition,a);
			if ( model.isPartitionable(a.getLength()) ) {
				tmp.add(a);
			}
			sortOrder.add(a);
		}
		
		if ( tmp.isEmpty() ) return null;

		for ( ITuple t : node.getSample() ) {
			for ( Axis a : tmp ) {
				a.update(t);
			}
		}
		
		Collections.sort(sortOrder,new Comparator<Axis>() {
			@Override
			public int compare(Axis o1, Axis o2) {
				return (int)Math.signum(o2.getLength() - o1.getLength());
			}});
		
		return new PreparedAxes(tmp,sortOrder);
	}
	
	public static final class HierarchicalChooser extends AxisChooser {
		@Override
		public Axis choose(PartitionNode p,Axis... axes) {
			Partition partition = p.getPartition();
			int index = partition.getLevel() % partition.getArity();
			for ( Axis axis : axes ) {
				if ( axis.getIndex() == index ) return axis;
			}
			return null;
		}
	}
	
	public static final class PeakChooser extends AxisChooser implements Comparator<Axis> {
		@Override
		public Axis choose(PartitionNode p,Axis... axes) {
			Arrays.sort(axes,this);
			return axes[0];
		}

		@Override
		public int compare(Axis o1, Axis o2) {
			return o2.getHistogram().getMaxFrequency() - o1.getHistogram().getMaxFrequency();
		}
	}
	
	public static final class MaxVarianceChooser extends AxisChooser implements Comparator<Axis> {
		@Override
		public Axis choose(PartitionNode p,Axis... axes) {
			Arrays.sort(axes,this);
			return axes[0];
		}

		@Override
		public int compare(Axis o1, Axis o2) {
			return (int)Math.signum(o2.getStdev() - o1.getStdev());
		}		
	}
	
	public static final class ConstChooser extends AxisChooser {
		private final int index;
		
		public ConstChooser(int i) {
			index = i;
		}
		
		@Override
		public Axis choose(PartitionNode p,Axis... axes) {
			for ( Axis axis : axes ) {
				if ( axis.getIndex() == index ) return axis;
			}
			return null;
		}
	}
}